package com.cupdata.zicon.conf;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

import javax.annotation.PostConstruct;

import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TMultiplexedProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransport;
import org.apache.thrift.transport.TTransportException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import com.cupdata.thrift.zicon.interfaces.*;

/**
 * 
* @Title: UserServiceZooKeeperConfig.java 
* @Description: 对注册在Zookeeper上的thrift服务进行监听
* @author dongliangwu
* @date 2017年2月14日
* @version V1.0
 */
public class ServiceZooKeeperConfig {
	
	/*@Value("${service.name}")
	String serviceName;

	@Value("${zookeeper.server.list}")
	String zookeeperList;
	
	@Value("${thread.pool.size}")
    int threadPoolSize;
	
	@Value("${time.to.sleep}")
	long timeToSleep;
	
	ExecutorService executor;
	
	TTransport transport;

	// thrift实例列表
	public static Map<String, UserService.Client> userServiceMap = new HashMap<String, UserService.Client>();
	public static Map<String, HelloService.Client> helloServiceMap = new HashMap<String, HelloService.Client>();

	@PostConstruct
	private void init() {
		executor = Executors.newFixedThreadPool(threadPoolSize);
		executor.execute(new Runnable() {
			@Override
			public void run() {
				startZooKeeper();
				try {
					TimeUnit.SECONDS.sleep(timeToSleep);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}
		});
	}

	// 服务监听
	private void startZooKeeper() {
		List<String> currChilds = new ArrayList<String>();
		String servicePath = "/" + serviceName;// 根节点路径
		ZkClient zkClient = new ZkClient(zookeeperList);
		boolean serviceExists = zkClient.exists(servicePath);
		if (serviceExists) {
			currChilds = zkClient.getChildren(servicePath);
		} else {
			throw new RuntimeException("service not exist!");
		}
		//与服务端建立连接
//		createInstance(String serviceInstanceName)
		for (String instanceName : currChilds) {
			// 没有该服务，建立该服务
			transport= createInstance(instanceName);
			if (!userServiceMap.containsKey(instanceName)) {
				userServiceMap.put(instanceName, new UserService.Client(new TMultiplexedProtocol(new TBinaryProtocol(transport),"UserService")));
			}
			if (!helloServiceMap.containsKey(instanceName)) {
				helloServiceMap.put(instanceName, new HelloService.Client(new TMultiplexedProtocol(new TBinaryProtocol(transport),"HelloService")));
			}
			
			
		}
		// 注册事件监听
		zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
			// @Override
			public void handleChildChange(String parentPath,
					List<String> currentChilds) throws Exception {
				// 实例(path)列表:当某个服务实例宕机，实例列表内会减去该实例
				for (String instanceName : currentChilds) {
					// 没有该服务，建立该服务
					if (!userServiceMap.containsKey(instanceName)) {
						userServiceMap.put(instanceName,new UserService.Client(new TMultiplexedProtocol(new TBinaryProtocol(transport),"UserService")));
					}
					if (!helloServiceMap.containsKey(instanceName)) {
						helloServiceMap.put(instanceName,new HelloService.Client(new TMultiplexedProtocol(new TBinaryProtocol(transport),"HelloService")));
					}
				}
				
				
				for (Map.Entry<String, UserService.Client> entry : userServiceMap.entrySet()) {
					// 该服务已被移除
					if (!currentChilds.contains(entry.getKey())) {
						UserService.Client c = userServiceMap.get(entry.getKey());
						try {
							c.getInputProtocol().getTransport().close();
							c.getOutputProtocol().getTransport().close();
						} catch (Exception e) {
							e.printStackTrace();
						}
						userServiceMap.remove(entry.getKey());
					}
				}
				
				for (Map.Entry<String, HelloService.Client> entry : helloServiceMap.entrySet()) {
					// 该服务已被移除
					if (!currentChilds.contains(entry.getKey())) {
						HelloService.Client c = helloServiceMap.get(entry.getKey());
						try {
							c.getInputProtocol().getTransport().close();
							c.getOutputProtocol().getTransport().close();
						} catch (Exception e) {
							e.printStackTrace();
						}
						helloServiceMap.remove(entry.getKey());
					}
				}
			}
		});
	}

	
	
	
	// 创建一个服务实例
//	private Map<String,Object> createServiceInstance(String serviceInstanceName) {
//		Map<String,Object> map = new HashMap<String,Object>();
//		
//		String ip = serviceInstanceName.split("-")[1];
//		int thriftServerPort = Integer.valueOf(serviceInstanceName.split("-")[2]);
//		TSocket transport = new TSocket(ip,thriftServerPort);
//		try {
//			transport.open();
//		} catch (TTransportException e) {
//			e.printStackTrace();
//		}
//		map.put("UserService", new UserService.Client(new TBinaryProtocol(transport)));
//		map.put("HelloService", new HelloService.Client(new TBinaryProtocol(transport)));
//		return map;
//	}
	
	*//**
	 * 获取一个服务实例对象
	 * @param serviceInstanceName
	 * @return
	 *//*
	private TTransport createInstance(String serviceInstanceName) {
		String ip = serviceInstanceName.split("-")[1];
		int thriftServerPort = Integer.valueOf(serviceInstanceName.split("-")[2]);
		TTransport transport = new TSocket(ip,thriftServerPort);
		try {
			transport.open();
		} catch (TTransportException e) {
			e.printStackTrace();
		}
		return transport;
	}*/

//	private HelloService.Client createHelloService(String serviceInstanceName) {
//		String ip = serviceInstanceName.split("-")[1];
//		int thriftServerPort = Integer.valueOf(serviceInstanceName.split("-")[2]);
//		TSocket transport = new TSocket(ip,thriftServerPort);
//		try {
//			transport.open();
//		} catch (TTransportException e) {
//			e.printStackTrace();
//		}
//		return new HelloService.Client(new TBinaryProtocol(transport));
//	}
	
	
}
